package net.bwie.jtp.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.gson.JsonObject;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;

/*
   将Flink CDC实时同步数据,提取核心字段,并且转换值,最后拼接为JSON字符串放回
 */
public class CdcDataEtlMapFunction extends RichMapFunction<String, String> {
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
    }

    /*
       1.解析JSON
       2.提取字段值
       3.拼接JSON字符串
       4. 返回
     */
    @Override
    public String map(String value) throws Exception {
        // TODO: 2025/6/2 解析JSON数据
        JSONObject jsonObject = JSON.parseObject(value);
        // TODO: 2025/6/2 提取字段
        // TODO: 2025/6/2 提取after字段
        String after = jsonObject.getString("after");
        // TODO: 2025/6/2 提取before字段
        String before = jsonObject.getString("before");
        // TODO: 2025/6/2 提取op字段
        String opValue = jsonObject.getString("op");
        // TODO: 2025/6/2 提取ts_ms字段
        Long ts_ms = jsonObject.getLong("ts_ms");
        // TODO: 2025/6/2 提取source字段
        JSONObject source = jsonObject.getJSONObject("source");
        // TODO: 2025/6/2 提取source里的db字段
        String db = source.getString("db");
        // TODO: 2025/6/2 提取source里的table字段
        String table = source.getString("table");
        // TODO: 2025/6/2 判断op字段的value
        String operatorValue;
        switch (opValue) {
            case "u":
                operatorValue = "update";
                break;
            case "r":
            case "c":
                operatorValue = "insert";
                break;
            case "d":
                operatorValue = "delete";
                break;
            default:
                operatorValue = "unknown";
        }
        // TODO: 2025/6/2 拼接字符串
        JSONObject newObject = new JSONObject();
        newObject.put("db_name", db);
        newObject.put("table_name", table);
        newObject.put("operate_type", operatorValue);
        newObject.put("ts_ms", ts_ms);
        if ("d".equals(opValue)) {
           newObject.put( "operate_data", JSON.parseObject(before));
        } else {
            newObject.put("operate_data", JSON.parseObject(after));
        }
        return newObject.toJSONString();
    }

    @Override
    public void close() throws Exception {
        super.close();
    }
}
